package com.idroc.coelacanth.core.dht;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;

import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.coelacanth.core.dht.NodeActionObserver.NodeAction;
import com.idroc.coelacanth.core.membership.Endpoint;
import com.idroc.coelacanth.core.membership.EndpointMeta;
import com.idroc.coelacanth.core.membership.Gossiper;
import com.idroc.coelacanth.core.membership.MemberState;
import com.idroc.coelacanth.core.membership.MemberStateObserver;
import com.idroc.coelacanth.core.message.MessageDispatchHandler;
import com.idroc.coelacanth.locate.support.ServiceLocators;
import com.idroc.coelacanth.service.ThreadPoolService;

/**
 * 一致性hash环
 * @author yuyoo (yuyoo4j@163.com)
 * @date Dec 12, 2009 3:38:57 PM
 */
public class ConsistentHashRing extends ServiceLocators.ConsistentHashServiceLocator {
	
	private static final Logger log = LoggerFactory.getLogger(ConsistentHashRing.class);
		
	/**
	 * 流言协议实现对象
	 */
	private Gossiper gossiper = null;
	
	/**
	 * 当前一致性hash节点
	 */
	private ConsistendHashNode currentNode = null;
	
	/**
	 * 当前一致性hash节点执行的动作
	 */
	private NodeAction currentNodeAction = NodeAction.NULL;
	
	/**
	 * 开放节点集合
	 */
	private final SortedMap<Integer, ConsistendHashNode> opens = new TreeMap<Integer, ConsistendHashNode>();
	
	/**
	 * 关闭节点集合
	 */
	private final SortedMap<Integer, ConsistendHashNode> closes = new TreeMap<Integer, ConsistendHashNode>();
	
	/**
	 * 种子节点集合
	 */
	private SortedMap<Integer, ConsistendHashNode> seeds = new TreeMap<Integer, ConsistendHashNode>();
	
	/**
	 * 端点<->节点映射关系
	 */
	private final Map<Endpoint, Integer> endpointNodeMapping = new ConcurrentHashMap<Endpoint, Integer>(100);
	
	/**
	 * 节点行为 观察者集合
	 */
	private final List<NodeActionObserver> observers = new ArrayList<NodeActionObserver>(100);
	 	
	/**
	 * 是否已经启动
	 */
	private boolean boot = false;
		
	public ConsistentHashRing(Gossiper gossipService, ConsistendHashNode node, NodeAction action) {
		
		this.gossiper = gossipService;
		this.gossiper.addObserver(new GossiperMemberStateOberser());
		this.currentNode = node;
		this.currentNodeAction = action;
		endpointNodeMapping.put(node.getEndpoint(), Integer.valueOf(node.getNo())); // 建立端点<->节点映射关系
	}
	
	/**
	 * 启动一致性hash环
	 */
	public void start() {
		
		registerMessageHandler(); // 注册消息处理器
		boot = true;
	}
	
	/**
	 * 关闭一致性hash环
	 */
	public void stop() {
		
		if (!boot) {
			throw new IllegalStateException("一致性hash环未启动");
		}
	}  
	
	/**
	 * 增加 节点动作观察者
	 * @param o -- 节点观察者
	 */
	public void addNodeActionObserver(NodeActionObserver o) {
		
		log.debug("增加 节点观察者[{}]", o);
		observers.add(o);
	}
	
	/**
	 * 获取 节点动作观察者列表
	 * @return -- 节点观察者列表
	 */
	public List<NodeActionObserver> getNodeActionObservers() {
		return observers;
	}
	
	/**
	 * 节点发起动作, 通知节点动作观察者
	 * @param action -- 节点动作
	 * @param node -- 节点
	 * @param ring -- 一致性hash环
	 */
	private void noticeNodeAction(final NodeAction action, final int nodeNo, final ConsistentHashRing ring) {
		 
		for (NodeActionObserver item : ring.getNodeActionObservers()) {
			try {
				final NodeActionObserver nao = item;
				Runnable task = new Runnable() {
					public void run() {
						nao.update(action, nodeNo, ring); 				
					}
				};
				ThreadPoolService.getMaintainThreadPool().submit(task);
			} catch (Exception ioex) {
				log.error("节点[{}]发起动作[{}]", nodeNo, action);
				log.error("通知节点动作观察者[{}]异常", item);
				log.error("通知异常", ioex);
			}
		}
	}
	
	/**
	 * 询问 节点将要发起什么动作
	 * @param node -- 节点
	 */
	private void questionNodeAction(int toNo, Endpoint to, Endpoint from) {
		
		NodeActionRequestMessage msg = new NodeActionRequestMessage();
		msg.setNodeNo(toNo);
		msg.setFrom(from);
		gossiper.sendMessageTask(to, msg);
		log.debug("发送 节点动作请求消息[{}]到端点[{}]", msg, to);
	}
	
	/**
	 * 注册消息处理器
	 */
	private void registerMessageHandler() {
		
		MessageDispatchHandler dispatcher = gossiper.getMsgService().getServerMessageFactory().getMessageDispatcher();
		// 处理"端点转节点请求消息"
		dispatcher.addUpMessageHandler(EndpointToNodeRequestMessage.class, new EndpointToNodeRequestMessageHandler());
		// 处理"端点节点响应消息"
		dispatcher.addUpMessageHandler(EndpointToNodeResponseMessage.class, new EndpointToNodeResponseMessageHandler());
		// 处理"节点动作请求消息"
		dispatcher.addUpMessageHandler(NodeActionRequestMessage.class, new NodeActionRequestMessageHandler());
		// 处理"节点动作响应消息"
		dispatcher.addUpMessageHandler(NodeActionResponseMessage.class, new NodeActionResponseMessageHandler());
	}
	
	@Override
	public Object lookup(final String key, int strategyCode) {

		if (!boot) {
			throw new IllegalStateException("一致性hash环未启动");
		} 
		Iterator<ConsistendHashNode> it = clockwiseNodeSequence(key);
		if (it.hasNext()) {
			return it.next();
		} else {
			return null;
		}
	}
	
	/**
	 * 按顺时针方向次序获取节点
	 * @param start -- 开始节点
	 * @param startInclude -- 开始节点是否包含在序列中
	 * @return -- 迭代器Iterator<ConsistendHashNode>
	 */
	public Iterator<ConsistendHashNode> clockwiseNodeSequence(int start, boolean startInclude) {
		
		int no = startInclude ? start : start + 1;
		ConsistendHashNodeClockwiseIterator it = new ConsistendHashNodeClockwiseIterator(no, opens, opens.size());
		return it;
	}
	
	/**
	 * 按顺时针方向次序获取节点
	 * @param key -- key
	 * @return -- 迭代器Iterator<ConsistendHashNode>
	 */
	public Iterator<ConsistendHashNode> clockwiseNodeSequence(final String key) {
		
		int no = h.hash(key);
		Iterator<ConsistendHashNode> it = new ConsistendHashNodeClockwiseIterator(no, opens, opens.size());
		return it;
	}
	
	/**
	 * 按逆时针方向次序获取节点
	 * @param start -- 开始节点 
	 * @param startInclude -- 开始节点是否包含在序列中
	 * @return -- 迭代器Iterator<ConsistendHashNode>
	 */
	public Iterator<ConsistendHashNode> counterclockwiseNodeSequence(int start, boolean startInclude) {

		int no = startInclude ? start : start - 1;
		Iterator<ConsistendHashNode> it = new ConsistendHashNodeCounterclockwiseIterator(no, opens, opens.size());
		return it;
	}
	
	/**
	 * 按逆时针方向次序获取节点
	 * @param key -- key
	 * @return -- 迭代器Iterator<ConsistendHashNode>
	 */
	public Iterator<ConsistendHashNode> counterclockwiseNodeSequence(final String key) {
		
		int no = h.hash(key);
		Iterator<ConsistendHashNode> it = new ConsistendHashNodeCounterclockwiseIterator(no, opens, opens.size());
		return it;
	}
	
	/**
	 * 获取 流言协议实现对象
	 * @return -- 流言协议实现对象(Gossiper)
	 */
	public Gossiper getGossiper() {
		return gossiper;
	}

	/**
	 * 获取当前一致性hash节点
	 * @return -- 节点
	 */
	public ConsistendHashNode getCurrentNode() {
		return currentNode;
	}

	/**
	 * 设置当前一致性hash节点
	 * @param currentNode -- 一致性hash节点
	 */
	public void setCurrentNode(ConsistendHashNode currentNode) {
		this.currentNode = currentNode;
	}

	/**
	 * 设置当前一致性hash节点 动作
	 * @return -- 节点 动作
	 */
	public NodeAction getCurrentNodeAction() {
		return currentNodeAction;
	}

	/**
	 * 设置当前一致性hash节点 动作
	 * @param currentNodeAction -- 节点 动作
	 */
	public void setCurrentNodeAction(NodeAction currentNodeAction) {
		this.currentNodeAction = currentNodeAction;
	}

	/**
	 * 获取开放节点集合
	 * @return -- 节点集合
	 */
	public SortedMap<Integer, ConsistendHashNode> getOpens() {
		return opens;
	}

	/**
	 * 获取 关闭节点集合
	 * @return -- 节点集合 
	 */
	public SortedMap<Integer, ConsistendHashNode> getCloses() {
		return closes;
	}

	/**
	 * 获取种子节点集合
	 * @return -- 节点集合 
	 */
	public SortedMap<Integer, ConsistendHashNode> getSeeds() {
		return seeds;
	}

	/**
	 * 获取 端点<->节点映射关系
	 * @return -- 端点<->节点映射关系Map
	 */
	public Map<Endpoint, Integer> getEndpointNodeMapping() {
		return endpointNodeMapping;
	}

	@Override
	public void setNodes(Map<Integer, Object> nodes) {

		throw new UnsupportedOperationException("一致性hash环不能设置节点集合");
	}

	/**
	 * 一致性hash环节点顺时针迭代器(Iterator)
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date Dec 13, 2009 11:13:00 AM
	 */
	public static class ConsistendHashNodeClockwiseIterator implements Iterator<ConsistendHashNode> {
		
		private SortedMap<Integer, ConsistendHashNode> nodes = null;
		private int primaryKey = -1;
		private int currentKey = -1;
		private int remaining = 0;
		
		private ConsistendHashNodeClockwiseIterator(int primaryKey, SortedMap<Integer, ConsistendHashNode> nodes, int remaining) {
			 
			this.primaryKey = primaryKey;
			this.currentKey = primaryKey;
			this.nodes = nodes;
			this.remaining = remaining;
		}

		public boolean hasNext() {
			
			return remaining > 0;
		}

		public ConsistendHashNode next() {
			
			remaining--;
			ConsistendHashNode n = nodes.get(currentKey);
			int nextKey = currentKey + 1;
			if (null == n) {
				SortedMap<Integer, ConsistendHashNode> tails = nodes.tailMap(currentKey);
				Integer firstKey = null;
				if (tails.isEmpty()) {
					firstKey = nodes.firstKey();
					n = nodes.get(firstKey); 
				} else {
					firstKey = tails.firstKey();
					n = nodes.get(firstKey);
				}
				nextKey = firstKey.intValue() + 1;
			} 
			currentKey = nextKey;
			return n;
		}

		public void remove() {
			throw new UnsupportedOperationException("不支持移除操作");
		}		
		
		@Override
		public String toString() {

			StringBuilder sb = new StringBuilder();
			sb.append("ConsistendHashNodeClockwiseIterator[");
			sb.append("primaryKey").append(primaryKey).append(",");
			sb.append("currentKey").append(currentKey).append(",");
			sb.append("remaining").append(remaining).append(",");
			sb.append("nodes").append(nodes);
			sb.append("]");
			return sb.toString();
		}
	}
	
	/**
	 * 一致性hash环节点逆时针迭代器(Iterator)
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date Dec 13, 2009 11:43:21 AM
	 */
	public static class ConsistendHashNodeCounterclockwiseIterator implements Iterator<ConsistendHashNode> {
		
		private SortedMap<Integer, ConsistendHashNode> nodes = null;
		private int primaryKey = -1;
		private int currentKey = -1;
		private int remaining = 0;
		
		private ConsistendHashNodeCounterclockwiseIterator(int primaryKey, SortedMap<Integer, ConsistendHashNode> nodes, int remaining) {
			 
			this.primaryKey = primaryKey;
			this.currentKey = primaryKey;
			this.nodes = nodes;
			this.remaining = remaining;
		}

		public boolean hasNext() {
			
			return remaining > 0;
		}

		public ConsistendHashNode next() {
			
			remaining--;
			ConsistendHashNode n = nodes.get(currentKey);
			int nextKey = currentKey - 1;
			if (null == n) {
				SortedMap<Integer, ConsistendHashNode> heads = nodes.headMap(Integer.valueOf(currentKey + 1));
				Integer lastKey = null;
				if (heads.isEmpty()) {
					lastKey = nodes.lastKey();
					n = nodes.get(lastKey); 
				} else {
					lastKey = heads.lastKey();
					n = nodes.get(lastKey);
				}
				nextKey = lastKey.intValue() - 1;
			} 
			currentKey = nextKey;
			return n;
		}

		public void remove() {
			throw new UnsupportedOperationException("不支持移除操作");
		}			
		
		@Override
		public String toString() {

			StringBuilder sb = new StringBuilder();
			sb.append("ConsistendHashNodeCounterclockwiseIterator[");
			sb.append("primaryKey").append(primaryKey).append(",");
			sb.append("currentKey").append(currentKey).append(",");
			sb.append("remaining").append(remaining).append(",");
			sb.append("nodes").append(nodes);
			sb.append("]");
			return sb.toString();
		}
	}
	
	/**
	 * gossiper 成员状态观察者
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date Dec 12, 2009 3:39:26 PM
	 */
	private class GossiperMemberStateOberser implements MemberStateObserver {
		
		private final Logger log = LoggerFactory.getLogger(this.getClass());

		public void update(MemberState state, EndpointMeta item) { 

			Integer itemNo = endpointNodeMapping.get(item.endpoint); // 获取端点<->节点映射关系中的节点
			
			switch (state) {
			
				case JOIN: // 加入状态
					if (item.endpoint.equals(currentNode.getEndpoint())) {
						
						SortedMap<Integer, EndpointMeta> openMetas = gossiper.getOpenMetas();
						for (Integer key : openMetas.keySet()) {
							try {
								EndpointMeta em = openMetas.get(key);
								if (!em.endpoint.equals(currentNode.getEndpoint())) {
									log.debug("向开放状态端点[{}]查询其节点编号", em.endpoint);
									EndpointToNodeRequestMessage msg = new EndpointToNodeRequestMessage();
									msg.setFromNo(currentNode.getNo());
									msg.setFrom(currentNode.getEndpoint());
									msg.setCandidate(em.endpoint);
									gossiper.sendMessage(em.endpoint, msg);			 
								} else { 
								}
							} catch (Exception ex) {
								log.error("向开放状态端点查询其节点编号异常");
							}
						}
					} else { // 其他节点信息
						log.debug("OTHER ENDPOINT JOIN->item[{}]", item); 
					}
					break;
				case OPEN: // 开放状态
					log.debug("OPEN->item[{}]", item); 
					if (!item.endpoint.equals(currentNode.getEndpoint())) {						 
						if (null != itemNo) { 
							questionNodeAction(itemNo, item.endpoint, currentNode.getEndpoint()); // 询问节点动作
						} else {
							log.error("在端点<->节点映射关系不存在, 端点元数据[{}]", item); 
						}
					} else {						
						log.debug("节点[{}]发起动作, 通知一致性hash环的节点动作观察者", currentNode);
						noticeNodeAction(currentNodeAction, currentNode.getNo(), ConsistentHashRing.this);
					}
					break;
				case CLOSE: // 关闭状态
					log.debug("CLOSE->item[{}]", item); 
					if (null != itemNo) {		
						ConsistendHashNode inode = opens.remove(itemNo); 
						closes.put(itemNo, inode);
						log.debug("节点[{}]转入关闭集合[{}]", inode, closes); 
						log.debug("节点[{}]发起关闭动作, 通知一致性hash环的节点动作观察者", inode);
						noticeNodeAction(NodeAction.CLOSE, itemNo, ConsistentHashRing.this);
					}
					break;
				default: // 未知状态
					log.debug("ELSE->item[{}]", item);
			} 
		} 
	} 
	
	/**
	 * 端点转节点请求消息 处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date Dec 12, 2009 4:46:58 PM
	 */
	private class EndpointToNodeRequestMessageHandler extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {
			
			EndpointToNodeRequestMessage msg = (EndpointToNodeRequestMessage)e.getMessage();
			log.debug("端点转节点请求消息处理器,消息:{}", msg);

			if (msg.getCandidate().equals(currentNode.getEndpoint())) {
				
				Integer nodeNo = Integer.valueOf(msg.getFromNo());
				
				// 建立映射关系 
				endpointNodeMapping.put(msg.getFrom(), nodeNo);
				
				EndpointToNodeResponseMessage res = new EndpointToNodeResponseMessage();
				res.setNo(currentNode.getNo());
				res.setNodeType(currentNode.getNodeType());
				res.setCandidate(currentNode.getEndpoint());
				if (null != opens.get(Integer.valueOf(currentNode.getNo()))) {
					res.setOpened(true);
				} else {
					res.setOpened(false);
				}
				log.debug("节点发送端点转节点响应消息");
				gossiper.sendMessage(msg.getFrom(), res);
			} else {
				log.error("消息中候选端点与当前节点[{}]的端点不匹配", currentNode);
			}
		}
	}
	
	/**
	 * 端点转节点响应消息 处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date Dec 12, 2009 5:11:03 PM
	 */
	private class EndpointToNodeResponseMessageHandler extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {

			EndpointToNodeResponseMessage msg = (EndpointToNodeResponseMessage)e.getMessage();
			log.debug("端点转节点响应消息 处理器,消息:{}", msg);
			Integer no = Integer.valueOf(msg.getNo());
			ConsistendHashNode node = ConsistendHashNodes.newNodeInstance(msg.getNodeType());
			node.setNo(msg.getNo()); 
			node.setEndpoint(msg.getCandidate());
			if (msg.isOpened()) {
				log.debug("节点[{}]加入开放集合[{}]", node, opens);
				opens.put(no, node);
			}
			endpointNodeMapping.put(msg.getCandidate(), no);
			log.debug("为节点建立端点<->节点映射关系[{}]", endpointNodeMapping);
		}
	}
	
	/**
	 * 节点动作请求消息 处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-14 下午02:29:58
	 */
	private class NodeActionRequestMessageHandler extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {

			NodeActionRequestMessage msg = (NodeActionRequestMessage)e.getMessage();
			log.debug("节点动作请求消息 处理器,消息:{}", msg); 
			
			if (msg.getNodeNo() == currentNode.getNo()) {
				NodeActionResponseMessage res = new NodeActionResponseMessage();
				res.setAction(currentNodeAction);
				res.setNodeNo(currentNode.getNo());
				res.setNodeType(currentNode.getNodeType());
				res.setFrom(currentNode.getEndpoint());
				gossiper.sendMessage(msg.getFrom(), res);
				log.debug("发送节点动作响应消息[{}]", res);
			} else {
				log.error("节点动作请求消息[{}]与节点[{}]不一致", msg, currentNode);
			}
		}
	}
	
	/**
	 * 节点动作响应消息 处理器
	 * @author yuyoo (yuyoo4j@163.com)
	 * @date 2009-12-14 下午02:34:27
	 */
	private class NodeActionResponseMessageHandler extends SimpleChannelUpstreamHandler {
		
		@Override
		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
				throws Exception {

			NodeActionResponseMessage msg = (NodeActionResponseMessage)e.getMessage();
			log.debug("节点动作响应消息 处理器,消息:{}", msg); 
			 
			Endpoint ep = msg.getFrom();
			Integer nodeNo = endpointNodeMapping.get(ep);
			
			ConsistendHashNode node = ConsistendHashNodes.newNodeInstance(msg.getNodeType());
			node.setNo(msg.getNodeNo());
			node.setEndpoint(ep); 
			
			seeds.put(nodeNo, node); 
			log.debug("已将节点加入种子节点集合[{}]", seeds);
			
			log.debug("节点[{}]发起动作, 通知一致性hash环的节点动作观察者", nodeNo);
			noticeNodeAction(msg.getAction(), nodeNo.intValue(), ConsistentHashRing.this);
			 
		}
	}
}
